/*
 * Copyright (C) 2018 xuexiangjys(xuexiangjys@163.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.xuexiang.xdlink.component.channel.impl;

import com.xuexiang.xdlink.component.channel.IDataReceiver;
import com.xuexiang.xdlink.component.channel.OnChannelListener;
import com.xuexiang.xdlink.logs.DCFLog;
import com.xuexiang.xdlink.util.Utils;

import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;

/**
 * 抽象的数据接收器
 *
 * @author xuexiang
 * @since 2018/8/14 下午10:46
 */
public abstract class AbstractDataReceiver extends Thread implements IDataReceiver {

    /**
     * 默认的每次读取数据的长度为4K
     */
    public final static int DEFAULT_READ_DATA_LENGTH = 4096;
    /**
     * 默认的每次读取数据的间隔为50ms
     */
    public final static long DEFAULT_READ_DATA_INTERVAL = 50L;

    private Socket mSocket;
    /**
     * 输入流
     */
    protected InputStream mInputStream;
    /**
     * 每次读取数据的数据块
     */
    protected byte[] mReadData;
    /**
     * 每次读取的数据长度
     */
    protected int mInAvailable = 0;
    /**
     * 数据接收器是否运行
     */
    protected volatile boolean mIsRunning = false;
    /**
     * 通信通道的监听器
     */
    private OnChannelListener mOnChannelListener;
    /**
     * 读取线程的间期，默认是50毫秒
     */
    private long mInterval = DEFAULT_READ_DATA_INTERVAL;

    //==========================构造方法===============================//

    /**
     * 构造函数
     *
     * @param listener 通信通道的监听器
     */
    public AbstractDataReceiver(OnChannelListener listener) {
        this(null, DEFAULT_READ_DATA_LENGTH, listener);
    }

    /**
     * 构造函数
     *
     * @param socket   建立连接获取的socket
     * @param listener 通信通道的监听器
     */
    public AbstractDataReceiver(Socket socket, OnChannelListener listener) {
        this(socket, DEFAULT_READ_DATA_LENGTH, listener);
    }

    /**
     * 构造函数
     *
     * @param readDataLength 每次读取数据的最大长度
     * @param listener       通信通道的监听器
     */
    public AbstractDataReceiver(int readDataLength, OnChannelListener listener) {
        this(null, readDataLength, listener);
    }

    /**
     * 构造函数
     *
     * @param socket         建立连接获取的socket
     * @param readDataLength 每次读取数据的最大长度
     * @param listener       通信通道的监听器
     */
    public AbstractDataReceiver(Socket socket, int readDataLength, OnChannelListener listener) {
        setPriority(Thread.MAX_PRIORITY);
        mSocket = socket;
        mReadData = new byte[readDataLength];
        mOnChannelListener = listener;
    }

    /**
     * 设置socket
     *
     * @param socket socket连接
     */
    protected AbstractDataReceiver setSocket(Socket socket) {
        mSocket = socket;
        return this;
    }

    /**
     * 设置每次读取数据的最大长度
     *
     * @param readDataLength 每次读取数据的最大长度
     */
    protected AbstractDataReceiver setReadDataLength(int readDataLength) {
        mReadData = new byte[readDataLength];
        return this;
    }

    /**
     * 设置数据接收的间期
     *
     * @param interval 数据接收的间期
     */
    protected AbstractDataReceiver setInterval(long interval) {
        mInterval = interval;
        return this;
    }

    //===========================运行==============================//

    @Override
    public void run() {
        try {
            onStart();
            while (isRunning()) {
                beforeAcceptData();
                try {
                    Thread.sleep(getAcceptInterval());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                acceptData();
            }
            onStop();
        } catch (Exception e) {
            e.printStackTrace();
            onDataReceiveInterrupted(e);
        }
    }

    /**
     * 数据接收异常中断
     *
     * @param e
     */
    protected void onDataReceiveInterrupted(Exception e) {
        DCFLog.e("Data receiving exception!", e);
    }

    /**
     * @return 是否正在接收数据
     */
    protected boolean isRunning() {
        return mIsRunning;
    }

    @Override
    public synchronized void startWork() {
        if (!isRunning()) {
            DCFLog.i(getChannel().toString() + " data receiver is now starting...");
            mIsRunning = true;
            super.start();
        }
    }


    /**
     * 开始运行接收数据线程【打开流】
     *
     * @throws IOException
     */
    protected void onStart() throws IOException {
        if (mSocket != null) {
            mInputStream = mSocket.getInputStream();
        }
    }

    /**
     * 停止运行接收数据线程
     *
     * @throws IOException
     */
    protected void onStop() throws Exception {
        DCFLog.i(getChannel().toString() + " data receiver has stopped...");
        close();
    }

    @Override
    public void beforeAcceptData() throws Exception {

    }

    @Override
    public long getAcceptInterval() {
        return mInterval;
    }

    @Override
    public void acceptData() throws Exception {
        if (mInputStream != null) {
            mInAvailable = mInputStream.available();
            if (mInAvailable > 0) {
                if (mInAvailable > mReadData.length) {
                    mInAvailable = mReadData.length;
                }
                mInAvailable = mInputStream.read(mReadData, 0, mInAvailable);
                onDataReceived(mReadData, mInAvailable);
            }
        }
    }

    @Override
    public void onDataReceived(byte[] data, int available) {
        if (mOnChannelListener != null) {
            mOnChannelListener.onDataReceived(getChannel(), data, available);
        }
    }

    @Override
    public void finishWork() {
        mIsRunning = false;
        interrupt();
        close();
    }

    /**
     * 关闭数据流
     */
    protected void close() {
        Utils.closeIO(mInputStream);
        mInputStream = null;
        Utils.closeIO(mSocket);
        mSocket = null;
    }

    public Socket getSocket() {
        return mSocket;
    }
}
